package sink;

import bean.TableProcess;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import util.RedisUtil;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.List;
import java.util.Locale;

public class MyPhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private Connection conn;
    private ValueState<Boolean> isFirstState;

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context ctx) throws Exception {
        //1. 来的数据如果是第一条,判断表是否存在,不存在就创建表
        judgeFirstValueAndCreateTable(value);
        //2. 如果不是第一条数据,或者是第一条数据但是表存在,插入表中
        write2Hbase(value);
        //3. 数据插入到hbase中,如果是更新原有数据,需要更新前置redis缓存,保证在查询redis的热点数据时也是更新后的数据
        delRedisOldValue(value, 1);
    }

    private void delRedisOldValue(Tuple2<JSONObject, TableProcess> value, int redisDatabase) {
        Jedis jedis = RedisUtil.getRedisClient();
        jedis.select(redisDatabase);
        jedis.del(value.f1.getSinkTable().toUpperCase() + ":" + value.f0.get("id"));//redis中即便要删除的key不存在,也只会返回0,表示1个都没删除成功,不会报错
        jedis.close();
    }

    private void write2Hbase(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        TableProcess tableProcess = value.f1;
        JSONObject data = value.f0;

        String sinkColumns = tableProcess.getSinkColumns();
        String[] strings = sinkColumns.split(",");
        StringBuilder upsertSql = new StringBuilder();
        //upsert into table(id,name) values(?,?);
        upsertSql.append("upsert into ")
            .append(tableProcess.getSinkTable())
            .append("(");

        upsertSql.append(sinkColumns);

        upsertSql.append(")values(");

        upsertSql.append(sinkColumns.replaceAll("[^,]+", "?")); //正则
        upsertSql.append(")");


        PreparedStatement ps = conn.prepareStatement(String.valueOf(upsertSql));
        System.out.println(data);
        for (int i = 0; i < strings.length; i++) {
            Object o = data.get(strings[i]);
            ps.setString(i + 1, o == null ? "" : o.toString());
        }
        ps.executeUpdate();
        conn.commit();
        ps.close();
    }

    private void judgeFirstValueAndCreateTable(Tuple2<JSONObject, TableProcess> value) throws IOException, SQLException {
        TableProcess tableProcess = value.f1;
        // 使用键控状态
        if (isFirstState.value() == null) {

            //拼接建表语句
            StringBuilder createTableSql = new StringBuilder();
            createTableSql
                .append("create table if not exists ")
                .append(tableProcess.getSinkTable())
                .append("(");
            createTableSql
                .append(tableProcess.getSinkColumns().replaceAll(",", " varchar,"));

            createTableSql
                .append(" varchar,")
                .append("constraint pk primary key(");
            createTableSql
                .append(tableProcess.getSinkPk() == null ? "id" : tableProcess.getSinkPk())
                .append("))")
                .append(tableProcess.getSinkExtend() == null ? "" : tableProcess.getSinkExtend());

            //1. 执行建表语句
            //1.1预执行
            PreparedStatement ps = conn.prepareStatement(createTableSql.toString());
            //1.2执行
            ps.execute();
            //1.3提交事务
            conn.commit();
            //1.4关闭资源
            ps.close();
            //2.更新键控状态
            isFirstState.update(true);
        }
    }

    @Override
    public void close() throws Exception {
        // 关闭连接,每个并行度调用一次
        if (conn != null) {
            conn.close();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //注意:已经keyBy过了
        //open下创建连接:每个并行度调用一次
        //1 加载驱动
        //2 配置phoenix url
        //3 获取连接对象
        Class.forName(Constant.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(Constant.PHOENIX_URL);
        isFirstState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirstState", Boolean.class));
    }
}
